package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SprakStreaming06_Transform {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("tarnsfrom").setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(3))



    //1.RDD中的方法称为->算子
    //2.SparkStream DStream中的方法称为 原语
    //3.scala中集合的方法称之为方法


    //DStream 并没有提供大量的方法使开发人员使用,eg:sort
    val socketData: ReceiverInputDStream[String] = ssc.socketTextStream("localhost",9999)

   // val mapData: DStream[String] = socketData.map(_*2)


    //transform转换：获取底层的RDD进行数据处理
    val mapDS: DStream[String] = socketData.transform(
      rdd => {
        rdd.map(_ * 2)
      }
    )
    mapDS.print()

    //mapData.print()




    ssc.start()
    ssc.awaitTermination()
  }

}
